controlplane/src/bin/db-cleanup.ts (2)

88-101: Remove @ts-ignore and simplify chunk logic.

The @ts-ignore suppresses a type error for a condition that's always false (MAX_DEGREE_OF_PARALLELISM === 1 when it's defined as 5). Consider removing the special case or making it configurable if the single-threaded path is needed.

 function chunkArray<T>(data: T[]): T[][] {
-  // @ts-ignore
-  if (MAX_DEGREE_OF_PARALLELISM === 1) {
-    return [data];
-  }
-
   const chunks: T[][] = [];
   const organizationsPerChunk = Math.ceil(ORGANIZATIONS_PER_BUCKET / MAX_DEGREE_OF_PARALLELISM);
   for (let i = 0; i < data.length; i += organizationsPerChunk) {
     chunks.push(data.slice(i, i + organizationsPerChunk));
   }
-
   return chunks;
 }

---

117-117: Consider using the pino logger consistently.

A pino logger is created on line 62, but the script uses console.log/console.error for output (lines 117, 134, 138, 141, 156, 189). For consistency with the rest of the codebase and better structured logging, consider using the pino logger throughout.

Also applies to: 138-138, 141-141, 156-156, 189-189

controlplane/src/core/workers/NotifyOrganizationDeletionQueuedWorker.ts (2)

86-87: Use getOrganizationAdmins instead of filtering all members.

OrganizationRepository has a getOrganizationAdmins method that directly returns admin members. This avoids loading RBAC data for all members just to filter.

-    const organizationMembers = await orgRepo.getMembers({ organizationID: org.id });
-    const orgAdmins = organizationMembers.filter((m) => m.rbac.isOrganizationAdmin);
+    const orgAdmins = await orgRepo.getOrganizationAdmins({ organizationID: org.id });

---

100-100: Avoid direct process.env access; pass webBaseUrl via configuration.

The worker accesses process.env.WEB_BASE_URL directly, which is inconsistent with other workers that receive configuration through their input options. This also makes testing harder.

+// In createNotifyOrganizationDeletionQueuedWorker input:
+  webBaseUrl: string;

 // In handler:
-        restoreLink: `${process.env.WEB_BASE_URL}/${org.slug}/settings`,
+        restoreLink: `${this.input.webBaseUrl}/${org.slug}/settings`,

controlplane/src/core/build-server.ts (1)

406-411: Pass webBaseUrl to the worker for consistency.

The worker uses process.env.WEB_BASE_URL directly, but opts.auth.webBaseUrl is available here. Pass it to maintain consistency with how other components receive configuration.

   createNotifyOrganizationDeletionQueuedWorker({
     redisConnection: fastify.redisForWorker,
     db: fastify.db,
     logger,
     mailer: mailerClient,
+    webBaseUrl: opts.auth.webBaseUrl,
   }),

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: JivusAyrus
Repo: wundergraph/cosmo PR: 2156
File: controlplane/src/core/repositories/SubgraphRepository.ts:1749-1751
Timestamp: 2025-08-29T10:28:04.846Z
Learning: In the controlplane codebase, authentication and authorization checks (including organization scoping) are handled at the service layer in files like unlinkSubgraph.ts before calling repository methods. Repository methods like unlinkSubgraph() in SubgraphRepository.ts can focus purely on data operations without redundant security checks.

controlplane/src/core/workers/NotifyOrganizationDeletionQueuedWorker.ts (2)

76-78: Throwing error when mailer not configured causes unnecessary retries.

When the mailer is not configured, throwing an error triggers the retry mechanism (6 attempts with exponential backoff). This will cause the job to fail repeatedly for approximately 1.3 hours before giving up, wasting resources.

Consider logging a warning and returning early instead:

     if (!this.input.mailer) {
-      throw new Error('Mailer service not configured');
+      this.input.logger.warn('Mailer service not configured, skipping notification');
+      return;
     }

---

87-101: Handle case where organization has no admins.

If orgAdmins is empty (line 87), the email will be sent with an empty receiverEmails array (line 95). Consider adding a guard to skip the email or log a warning:

     const orgAdmins = organizationMembers.filter((m) => m.rbac.isOrganizationAdmin);
+
+    if (orgAdmins.length === 0) {
+      this.input.logger.warn({ organizationId: org.id }, 'No admins found for organization, skipping notification');
+      return;
+    }

     const intl = Intl.DateTimeFormat(undefined, {

controlplane/src/core/workers/ReactivateOrganizationWorker.ts (1)

115-116: LGTM! Typo fix aligns logging with other workers.

The change from joinId to jobId corrects a typo and ensures consistent logging across workers.

Optional: Rename the parameter for clarity

The job parameter in the stalled event callback is actually the job ID string, not a Job object. Consider renaming it to jobId for clarity:

-  worker.on('stalled', (job) => {
-    log.warn({ jobId: job }, `Job stalled`);
+  worker.on('stalled', (jobId) => {
+    log.warn({ jobId }, `Job stalled`);
   });

Based on learnings, this change aligns with similar updates across other workers in the PR.

controlplane/src/core/workers/DeactivateOrganizationWorker.ts (1)

127-129: LGTM! Key name fix improves consistency.

The change from joinId to jobId correctly aligns the log key with the actual value being logged and standardizes the approach across workers.

Optional: Rename parameter for clarity

The callback parameter job receives the jobId string (per BullMQ's stalled event signature), not a Job object. Consider renaming it to jobId for clarity:

-  worker.on('stalled', (job) => {
-    log.warn({ jobId: job }, `Job stalled`);
+  worker.on('stalled', (jobId) => {
+    log.warn({ jobId }, `Job stalled`);
   });

controlplane/src/core/workers/CacheWarmerWorker.ts (1)

133-135: Good typo fix; consider logging just the job ID for consistency.

The key name correction from joinId to jobId is excellent and aligns with the broader pattern across workers.

For consistency with the error handler on line 112 (which logs jobId: job.id), consider logging just the job ID rather than the entire job object. This keeps logs concise and matches the established pattern in this file.

🔎 Optional refinement for consistency:

-  log.warn({ jobId: job }, `Job stalled`);
+  log.warn({ jobId: job.id }, `Job stalled`);

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: JivusAyrus
Repo: wundergraph/cosmo PR: 2156
File: controlplane/src/core/repositories/SubgraphRepository.ts:1749-1751
Timestamp: 2025-08-29T10:28:04.846Z
Learning: In the controlplane codebase, authentication and authorization checks (including organization scoping) are handled at the service layer in files like unlinkSubgraph.ts before calling repository methods. Repository methods like unlinkSubgraph() in SubgraphRepository.ts can focus purely on data operations without redundant security checks.

Learnt from: wilsonrivera
Repo: wundergraph/cosmo PR: 1919
File: controlplane/src/core/repositories/OrganizationGroupRepository.ts:193-224
Timestamp: 2025-07-01T13:53:54.146Z
Learning: In the Cosmo codebase, database transactions are typically managed at the service layer (e.g., in buf services like deleteOrganizationGroup.ts), where repositories are instantiated with the transaction handle and all operations within those repositories are automatically part of the same transaction.

controlplane/src/bin/db-cleanup.ts (1)

149-161: Critical: Nested transaction and Redis atomicity issues remain unaddressed.

This is the same issue previously flagged: the outer transaction at line 151 wraps processChunkOfOrganizations, but within that function, orgRepo.queueOrganizationDeletion (line 199) starts its own transaction, creating nested transactions. Additionally, the Redis notification job (lines 207-211) is enqueued outside the DB transaction boundary, risking inconsistency if the transaction fails after the job is enqueued.

Required fixes:

1. Refactor orgRepo.queueOrganizationDeletion to accept and use the transaction context (tx) passed from the outer transaction, avoiding nested transactions
2. Move Redis job enqueueing inside the repository method or implement an outbox pattern to ensure the notification is only sent if the DB transaction commits successfully

Run the following script to verify the transaction handling in OrganizationRepository.queueOrganizationDeletion:

#!/bin/bash
# Check if queueOrganizationDeletion starts its own transaction
ast-grep --pattern $'queueOrganizationDeletion($$$) {
  $$$
  transaction($$$)
  $$$
}'

controlplane/src/bin/db-cleanup.ts (1)

56-72: Consider removing unused redisWorker connection.

The script connects and pings both redisQueue and redisWorker, but only redisQueue is used to initialize the queue instances. The redisWorker connection appears unnecessary for this script.

🔎 Proposed simplification

-  const { redisQueue, redisWorker } = await createRedisConnections({
+  const { redisQueue } = await createRedisConnections({
     host: redis.host!,
     port: Number(redis.port),
     password: redis.password,
     tls: redis.tls,
   });
 
   await redisQueue.connect();
-  await redisWorker.connect();
-  await redisWorker.ping();
   await redisQueue.ping();
 
   // ... rest of code ...
 
   } finally {
     redisQueue.disconnect();
-    redisWorker.disconnect();

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: JivusAyrus
Repo: wundergraph/cosmo PR: 2156
File: controlplane/src/core/repositories/SubgraphRepository.ts:1749-1751
Timestamp: 2025-08-29T10:28:04.846Z
Learning: In the controlplane codebase, authentication and authorization checks (including organization scoping) are handled at the service layer in files like unlinkSubgraph.ts before calling repository methods. Repository methods like unlinkSubgraph() in SubgraphRepository.ts can focus purely on data operations without redundant security checks.

Learnt from: wilsonrivera
Repo: wundergraph/cosmo PR: 1919
File: controlplane/src/core/repositories/OrganizationGroupRepository.ts:193-224
Timestamp: 2025-07-01T13:53:54.146Z
Learning: In the Cosmo codebase, database transactions are typically managed at the service layer (e.g., in buf services like deleteOrganizationGroup.ts), where repositories are instantiated with the transaction handle and all operations within those repositories are automatically part of the same transaction.

controlplane/src/bin/db-cleanup.ts (2)

122-122: Remove unused userId field from query.

The userId field is selected but never used in processChunkOfOrganizations. Consider removing it to slightly reduce query overhead.

🔎 Proposed fix

     .select({
       id: schema.organizations.id,
       slug: schema.organizations.slug,
-      userId: schema.organizations.createdBy,
       plan: schema.organizationBilling.plan,
     })

And update the type signature on line 173:

-  organizations: { id: string; slug: string; userId: string | null }[];
+  organizations: { id: string; slug: string }[];

---

182-182: Reuse the parent logger for consistency.

Creating a new pino() logger here instead of reusing the logger from line 62 leads to inconsistent logging across the script. The queue constructors (lines 67-68) receive the parent logger, but the repository receives a separate instance.

🔎 Proposed fix

Pass the logger as a parameter to processChunkOfOrganizations:

 async function processChunkOfOrganizations({
   organizations,
   db,
   inactivityThreshold,
   deleteOrganizationQueue,
   notifyOrganizationDeletionQueuedQueue,
+  logger,
 }: {
   organizations: { id: string; slug: string; userId: string | null }[];
   db: PostgresJsDatabase<typeof schema>;
   inactivityThreshold: Date;
   deleteOrganizationQueue: DeleteOrganizationQueue;
   notifyOrganizationDeletionQueuedQueue: NotifyOrganizationDeletionQueuedQueue;
+  logger: pino.Logger;
 }) {
   const queuedAt = new Date();
   const deletesAt = addDays(queuedAt, DELAY_FOR_ORG_DELETION_IN_DAYS);

-  const orgRepo = new OrganizationRepository(pino(), db, undefined);
+  const orgRepo = new OrganizationRepository(logger, db, undefined);

And update the call site around line 152:

         return processChunkOfOrganizations({
           organizations: chunk,
           db: tx,
           inactivityThreshold,
           deleteOrganizationQueue,
           notifyOrganizationDeletionQueuedQueue,
+          logger,
         });

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Learnt from: JivusAyrus
Repo: wundergraph/cosmo PR: 2156
File: controlplane/src/core/repositories/SubgraphRepository.ts:1749-1751
Timestamp: 2025-08-29T10:28:04.846Z
Learning: In the controlplane codebase, authentication and authorization checks (including organization scoping) are handled at the service layer in files like unlinkSubgraph.ts before calling repository methods. Repository methods like unlinkSubgraph() in SubgraphRepository.ts can focus purely on data operations without redundant security checks.

Learnt from: wilsonrivera
Repo: wundergraph/cosmo PR: 1919
File: controlplane/src/core/repositories/OrganizationGroupRepository.ts:193-224
Timestamp: 2025-07-01T13:53:54.146Z
Learning: In the Cosmo codebase, database transactions are typically managed at the service layer (e.g., in buf services like deleteOrganizationGroup.ts), where repositories are instantiated with the transaction handle and all operations within those repositories are automatically part of the same transaction.

controlplane/src/core/workers/QueueInactiveOrganizationsDeletionWorker.ts (2)

22-22: Convert empty interface to a type alias.

Per static analysis, an empty interface is equivalent to {}. Use a type alias for clarity and to follow best practices.

🔎 Proposed fix

-export interface QueueInactiveOrganizationsDeletionInput {}
+export type QueueInactiveOrganizationsDeletionInput = Record<string, never>;

---

215-222: Consider reducing concurrency for scheduled job.

This worker processes a single scheduled job hourly. A concurrency of 10 is excessive and won't provide any benefit. Consider setting concurrency: 1 to match the expected workload.

🔎 Proposed fix

   const worker = new Worker<QueueInactiveOrganizationsDeletionInput>(
     QueueName,
     (job) => new QueueInactiveOrganizationsDeletionWorker(input).handler(job),
     {
       connection: input.redisConnection,
-      concurrency: 10,
+      concurrency: 1,
     },
   );

controlplane/src/core/workers/NotifyOrganizationDeletionQueuedWorker.ts (2)

105-105: Avoid direct process.env access; inject via config.

Accessing process.env.WEB_BASE_URL directly breaks the configuration injection pattern used elsewhere. Consider passing webBaseUrl through the worker's input options for consistency and testability.

---

94-97: Specify a consistent locale for date formatting.

Using Intl.DateTimeFormat(undefined, ...) relies on the system's default locale, which can produce inconsistent date formats across environments. Consider using a fixed locale (e.g., 'en-US') or allowing the locale to be configured.

🔎 Proposed fix

-      const intl = Intl.DateTimeFormat(undefined, {
+      const intl = Intl.DateTimeFormat('en-US', {
         dateStyle: 'medium',
         timeStyle: 'short',
       });

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes